package com.gyj.register;

import com.gyj.register.entity.Applications;
import com.gyj.register.entity.DeltaRegistry;
import com.gyj.register.entity.RecentlyChangedServiceInstance;
import com.gyj.register.entity.ServiceInstance;
import com.gyj.register.enums.OperationTypeEnum;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicStampedReference;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 服务注册中心客户端缓存的一个服务注册表
 * @author Gao
 * @date 2021/6/26
 */
public class CachedServiceRegistry {

    private static final Logger logger = LogManager.getLogger(CachedServiceRegistry.class);

    /**
     * 服务注册表拉取间隔时间
     */
    private static final long SERVICE_REGISTRY_FETCH_INTERVAL = 30 * 1000L;

    /**
     * 客户端缓存的服务注册表
     * 多线程并发读，会有线程安全问题
     */
    ///private Map<String, Map<String, ServiceInstance>> registry = new HashMap<>();
    /**
     * AtomicReference类来保证引用对象之间的原子性，你可以把多个变量放在一个对象里来进行CAS操作
     * 多个地方多个线程同时对一个对象变量的引用进行赋值的时候，可能导致的并发冲突的问题，
     * 就用AtomicReference的CAS操作来解决了，而没有使用加锁的重量级的方式
     * 这里的全量拉取注册表和增量拉取都可能同时对注册表进行赋值，避免线程冲突
     *
     * eureka源码也是用的 AtomicReference缓存注册表新
     * AtomicReference<Applications> applicationsDelta = new AtomicReference<Applications>(new Applications());
     * 封装EurekaHttpClient也是用 AtomicReference<EurekaHttpClient>
     */
    ///private AtomicReference<Applications> applications = new AtomicReference<>(new Applications());
    /**
     * 解决ABA问题
     */
    private AtomicStampedReference<Applications> applications;

    /**
     * 代表了当前的本地缓存的服务注册表的一个版本号
     */
    private AtomicLong applicationsVersion = new AtomicLong(0L);

    /**
     * 客户端组件
     */
    private RegisterClient registerClient;

    /**
     * 发送通信的组件
     */
    private HttpSender httpSender;

    /**
     * 增量拉取注册表的组件
     */
    private FetchDeltaRegistryWorker fetchDeltaRegistryWorker;

    /**
     * 服务注册表的读写锁
     */
    private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    private ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
    private ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();

    /**
     * 构造函数初始化
     * @param registerClient
     * @param httpSender
     */
    public CachedServiceRegistry(RegisterClient registerClient, HttpSender httpSender) {
        // 初始化增量拉取注册表的工作线程
        this.fetchDeltaRegistryWorker = new FetchDeltaRegistryWorker();
        // 初始化客户端
        this.registerClient = registerClient;
        // 初始化网络通信组件
        this.httpSender = httpSender;
        // 初始化原子类修饰的注册表对象
        this.applications = new AtomicStampedReference<>(new Applications(), 0);
    }

    /**
     * 初始化定时任务
     */
    public void initialize() {
        // 启动全量拉取注册表的线程
        FetchFullRegistryWorker fetchFullRegistryWorker = new FetchFullRegistryWorker();
        fetchFullRegistryWorker.start();
        // 启动增量拉取注册表的线程
        this.fetchDeltaRegistryWorker.start();
    }

    /**
     * 销毁这个线程
     */
    public void destroy() {
        this.fetchDeltaRegistryWorker.interrupt();
    }

    /**
     * 增量拉取注册表的后台线程
     */
    private class FetchDeltaRegistryWorker extends Thread {
        @Override
        public void run() {
            while(registerClient.isRunning()) {
                try {
                    Thread.sleep(SERVICE_REGISTRY_FETCH_INTERVAL);

                    // 拉取回来的是最近3分钟变化的服务实例
                    DeltaRegistry deltaRegistry = httpSender.fetchDeltaRegistry();

                    // 一类是注册，一类是删除
                    // 如果是注册的话，就判断一下这个服务实例是否在这个本地缓存的注册表中
                    // 如果不在的话，就放到本地缓存注册表里去
                    // 如果是删除的话，就看一下，如果服务实例存在，就给删除了
                    // 我们这里其实是要大量的修改本地缓存的注册表，所以此处需要加锁
                    mergeDeltaRegistry(deltaRegistry);

                    // 再检查一下，跟服务端的注册表的服务实例的数量相比，是否是一致的
                    // 封装一下增量注册表的对象，也就是拉取增量注册表的时候，一方面是返回那个数据
                    // 另外一方面，是要那个对应的register-server端的服务实例的数量
                    compareRegistry(deltaRegistry);
                } catch (Exception e) {
                    logger.error("增量拉取注册表的后台线程，异常e:{}", e);
                }
            }
        }

        /**
         * 合并增量注册表到本地缓存注册表里去
         * @param deltaRegistry
         */
        private void mergeDeltaRegistry(DeltaRegistry deltaRegistry) {
            // 对注册表加锁
            try {
                writeLock.lock();
                // 获取全量注册表
                ///Map<String, Map<String, ServiceInstance>> registry = applications.get().getRegistry();
                Map<String, Map<String, ServiceInstance>> registry = applications.getReference().getRegistry();
                logger.info("增量拉取注册表，合并前的registry:{}", registry);
                ConcurrentLinkedQueue<RecentlyChangedServiceInstance> recentlyChangedQueue = deltaRegistry.getRecentlyChangedQueue();
                // 遍历变化的服务实例
                for(RecentlyChangedServiceInstance recentlyChangedItem : recentlyChangedQueue) {
                    ServiceInstance serviceInstance = recentlyChangedItem.getServiceInstance();
                    // 获取变化服务的服务器名称和服务实例id
                    String serviceName = serviceInstance.getServiceName();
                    String serviceInstanceId = serviceInstance.getServiceInstanceId();
                    // 如果是注册操作的话
                    if(Objects.equals(OperationTypeEnum.REGISTER.getCode(), recentlyChangedItem.getOperationType())) {
                        Map<String, ServiceInstance> serviceInstanceMap = registry.get(serviceName);
                        if(serviceInstanceMap == null) {
                            serviceInstanceMap = new HashMap<>(16);
                            registry.put(serviceName, serviceInstanceMap);
                        }
                        ServiceInstance serviceInstance1 = serviceInstanceMap.get(serviceInstanceId);
                        if(serviceInstance1 == null) {
                            serviceInstanceMap.put(serviceInstanceId, serviceInstance);
                        }
                        logger.info("增量拉取注册表，合并增量注册表到本地缓存注册表，注册操作，serviceInstanceMap:{}", serviceInstanceMap);
                    }
                    // 如果是删除操作的话
                    else if(OperationTypeEnum.REMOVE.getCode().equals(recentlyChangedItem.getOperationType())) {
                        Map<String, ServiceInstance> serviceInstanceMap = registry.get(serviceName);
                        if(serviceInstanceMap != null) {
                            serviceInstanceMap.remove(serviceInstanceId);
                            logger.info("增量拉取注册表, 操作是删除服务实例id:{}", serviceInstanceId);
                        }
                    }
                }
                logger.info("增量拉取注册表，合并后的registry:{}", registry);
            } finally {
                writeLock.unlock();
            }
        }

        /**
         * 校对注册表
         * @param deltaRegistry
         */
        private void compareRegistry(DeltaRegistry deltaRegistry) {
            // 本地缓存的注册表
            ///Map<String, Map<String, ServiceInstance>> registry = applications.get().getRegistry();
            Map<String, Map<String, ServiceInstance>> registry = applications.getReference().getRegistry();
            // 增量拉取的服务端服务实例总数
            long serverSideTotalCount = deltaRegistry.getServiceInstanceTotalCount();
            long clientSideTotalCount = 0L;
            for(Map<String, ServiceInstance> serviceInstanceMap : registry.values()) {
                clientSideTotalCount += serviceInstanceMap.size();
            }
            logger.info("增量拉取注册表, 本地计算后服务实例数量:{}, 服务端传过来的总的服务实例数量:{}",
                    clientSideTotalCount, serverSideTotalCount);
            // 如果拉取后的服务实例的数量和服务端传过来的总的实例数量不相等，则需要全量拉取注册表
            if(!Objects.equals(serverSideTotalCount, clientSideTotalCount)) {
                logger.info("重新拉取全量注册表进行纠正");
                // 人家正常的进行了全量注册表最新数据的一个赋值，可能是包含了40个服务实例, 最新数据
                fetchFullRegistry();
            }
        }
    }

    /**
     * 全量拉取注册表的后台线程
     */
 /*   private class FetchFullRegistryWorker extends Thread {
        @Override
        public void run() {
            // 拉取全量注册表, 服务启动的时候拉取一次
            Applications fetchedApplications = httpSender.fetchFullRegistry();
            while (true) {
                // 原有的缓存表
                Applications expectedApplications = CachedServiceRegistry.this.applications.get();
                // 设置拉取后的注册表赋值给本地缓存表
                if (applications.compareAndSet(expectedApplications, fetchedApplications)) {
                    break;
                }
            }
        }
    }*/

    /**
     * 全量拉取注册表的后台线程
     */
    private class FetchFullRegistryWorker extends Thread {
        @Override
        public void run() {
            // 拉取全量注册表
            // 这个操作要走网络，但是不知道为什么抽风了，此时就是一直卡住，数据没有返回回来
            // 卡在这儿了，卡了几分钟
            // 此时的这个数据是一个旧的版本，里面仅仅包含了30个服务实例
            // 全量拉注册表的线程突然苏醒过来了，此时将
            // 30个服务实例的旧版本的数据赋值给了本地缓存注册表
            fetchFullRegistry();
        }
    }

    /**
     * 拉取全量注册表到本地
     */
    private void fetchFullRegistry() {
        // 一定要在发起网络请求之前，先拿到一个当时的版本号
        // 接着在这里发起网络请求，此时可能会有别的线程来修改这个注册表，更新版本，在这个期间
        // 必须是发起网络请求之后，这个注册表的版本没有被人修改过，此时他才能去修改
        // 如果在这个期间，有人修改过注册表，版本不一样了，此时就直接if不成立，不要把你拉取到的
        // 旧版本的注册表给设置进去
        // version = 0
        Long expectedVersion = applicationsVersion.get();
        Applications fetchedApplications = httpSender.fetchFullRegistry();
        // version = 1, 如果此时版本号是1则这里不执行修改的逻辑，防止增量注册表的旧版本覆盖
        if(applicationsVersion.compareAndSet(expectedVersion, expectedVersion + 1)) {
            while(true) {
                Applications expectedApplications = applications.getReference();
                // 期望的邮戳，就是希望没有人修改过，不变，没人修改过则原子修改CAS操作
                int expectedStamp = applications.getStamp();
                if(applications.compareAndSet(expectedApplications, fetchedApplications,
                        expectedStamp, expectedStamp + 1)) {
                    break;
                }
            }
        }
    }

    /**
     * 获取服务注册表
     * 加读写锁，保证修改注册表全量拉取增量拉取或者是合并的时候
     * 不能读取数据
     * @return
     */
    /*public Map<String, Map<String, ServiceInstance>> getRegistry() {
        return applications.get().getRegistry();
    }*/
    public Map<String, Map<String, ServiceInstance>> getRegistry() {
        try {
            readLock.lock();
            return applications.getReference().getRegistry();
        } finally {
            readLock.unlock();
        }
    }

}
